package com.mq.rocketmq.normal;

import com.mq.rocketmq.pojo.DelayParam;
import com.mq.rocketmq.pojo.NormalEntity;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.*;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @Author: 19239
 * @Description:
 * @Date: 2024/6/27
 * @Version: 1.0
 */
@RestController
@RequestMapping("/send")
@Slf4j
public class Producer {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 发送普通消息
     * @param param
     */
    @PostMapping("/normal")
    public void send(@RequestBody NormalEntity param) {
        log.info("param: {},{},{}", param.getTopic(), param.getTags(), param.getMessage());
        Message<String> msg = MessageBuilder.withPayload(param.getMessage().toString())
                .setHeader(RocketMQHeaders.TAGS, param.getTags())
                .build();
        rocketMQTemplate.convertAndSend(param.getTopic(), msg);
    }

    /**
     * 发送延迟消息
     *
     * messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
     * <p>
     * delayLevel=1 表示消息延迟1秒。
     * delayLevel=2 表示消息延迟5秒。
     * delayLevel=18 表示消息延迟2小时。
     *
     * 值为0 不延迟
     *
     */

    @PostMapping("/delay")
    public void sendDelayedMessage(@RequestBody DelayParam param) {
        Message<String> msg = MessageBuilder.withPayload(param.getMessage().toString())
                .setHeader(MessageConst.PROPERTY_TAGS, param.getTags())
                .setHeader(MessageConst.PROPERTY_DELAY_TIME_LEVEL, String.valueOf(param.getDelayLevel()))
                .build();
        log.info("delay send time:{}",new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
        rocketMQTemplate.syncSend(param.getTopic(), msg,1*1000l,param.getDelayLevel());
    }

    /**
     * 发送同步消息（Sync Message），同步消息：它指的是生产者在发送一条消息后，会阻塞等待Broker的响应（ACK确认）。只有当收到Broker确认消息已成功存储或明确的失败反馈时，生产者才会继续执行后续操作
     * 可靠性：提供了较高的消息发送可靠性，因为生产者可以直接从Broker的响应中得知消息是否发送成功。
     * 适用于那些对消息丢失容忍度很低的场景，如金融交易、订单确认等重要业务流程中，确保消息至少一次的传递是必要的
     * @param param
     */

    @PostMapping("/sync")
    public void sendSyncMessage(@RequestBody NormalEntity param){
        log.info("sync send message time:{}",new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
        Message<String> msg = MessageBuilder.withPayload(param.getMessage().toString())
                .setHeader(MessageConst.PROPERTY_TAGS, param.getTags())
                .build();
        rocketMQTemplate.syncSend(param.getTopic(),msg);
    }


    /**
     * 发送异步消息（通过线程池执行发送到broker的消息任务，执行完后回调：在SendCallback中可处理相关成功失败时的逻辑）
     * （适合对响应时间敏感的业务场景）
     */
    @PostMapping("/asyn")
    public void sendAsyncMsg(@RequestBody NormalEntity param) {
        rocketMQTemplate.asyncSend(param.getTopic(), MessageBuilder.withPayload(param.getMessage()).build(), new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                // 处理消息发送成功逻辑
                log.info(" asyn sunccessed");
            }
            @Override
            public void onException(Throwable throwable) {
                // 处理消息发送异常逻辑
                log.info(" asyn failed");
            }
        });
    }

    /**
     * 发送单向消息（只负责发送消息，不等待应答，不关心发送结果，如日志）
     */
    @PostMapping("/sendOneWay")
    public void sendOneWayMsg(@RequestBody NormalEntity param) {
        rocketMQTemplate.sendOneWay(param.getTopic(), MessageBuilder.withPayload(param.getMessage()).build());
    }

    /**
     * 发送带tag的消息，直接在topic后面加上":tag"
     */
    @PostMapping("/tag")
    public SendResult sendTagMsg(@RequestBody NormalEntity param) {
        return rocketMQTemplate.syncSend(param.getTopic() + ":"+param.getTags()[0], MessageBuilder.withPayload(param.getMessage()).build());
    }


    public void sendOrderedStringMessage(@RequestBody NormalEntity param) {
        String topic = "ordered_string_message_topic";
        String message = "我是一条同步顺序消息:";
        for (int i = 0; i < 5; i++) {
            // hashkey是为了确保这些消息被路由到同一个消息队列，这样消费者就能够按照顺序处理它们
            rocketMQTemplate.syncSendOrderly(param.getTopic(), message + i, "alian_sync_ordered");
        }
    }

}
